package com.yifeng.repo.flink.data.transport.bootstrap;

import com.yifeng.repo.flink.data.transport.config.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.utils.ParameterTool;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 源为Kafka的启动helper类
 * @author wangzhi
 * @since 2023-06-08
 */
public class BootstrapSourceKafkaHelper extends BootstrapHelper {
    // 设置环境变量
    public static JobParametersConfig setEnvAndParameter(String[] args) throws IOException {
        JobParametersConfig config = new JobParametersConfig();
        ParameterTool tool = ParameterTool.fromArgs(args);
        String taskConfUrl = tool.get("task_config_path","");
        ParameterTool taskParas = ParameterTool.fromPropertiesFile(taskConfUrl);
        config.setCdcTaskname(taskParas.get("cdc.taskname"));
        config.setSourceParallelism(taskParas.getInt("source.parallelism"));
        config.setSinkParallelism(taskParas.getInt("sink.parallelism"));

        config.setStreamExecutionConfig(buildStreamExecutionConfig(taskParas, taskParas));

        KafkaConfig kafkaConfig = buildKafkaConfig(taskParas, taskParas);
        kafkaConfig.setGroupId(taskParas.get("kafka.groupId"));
        kafkaConfig.setStartingOffset(taskParas.get("kafka.startingOffset"));
        config.setKafkaConfig(kafkaConfig);

        config.setSinkIcebergConfig(buildSinkIcebergConfig(taskParas));
        return config;
    }
    /**
     * 设置sink输出到iceberg的配置
     * @param taskParas
     * @return
     */
    private static SinkIcebergConfig buildSinkIcebergConfig(ParameterTool taskParas){
        SinkIcebergConfig sinkIcebergConfig = new SinkIcebergConfig();
        sinkIcebergConfig.setCatalogWarehouse(taskParas.get("iceberg.catalog.warehouse"));
        sinkIcebergConfig.setCatalogUri(taskParas.get("iceberg.catalog.uri"));
        sinkIcebergConfig.setCatalogImpl(taskParas.get("iceberg.catalog.impl"));
        sinkIcebergConfig.setCatalogDatabase(taskParas.get("iceberg.catalog.database"));
        List<String> tableNameList = loadIcebergTableNames(taskParas.get("iceberg.catalog.tables"));
        sinkIcebergConfig.setCatalogTables(tableNameList);
        sinkIcebergConfig.setTableList(loadIcebergTables(tableNameList, taskParas));
        return sinkIcebergConfig;
    }

    /**
     * 获取iceberg的表名
     * @param tableNameStr
     * @return
     */
    private static List<String> loadIcebergTableNames(String tableNameStr) {
        if(org.apache.commons.lang3.StringUtils.isBlank(tableNameStr)){
            return null;
        }
        String[] str = org.apache.commons.lang3.StringUtils.split(tableNameStr,",");
        List<String> tableNameList = new ArrayList<>(str.length);
        Collections.addAll(tableNameList,str);
        return tableNameList;
    }

    /**
     * 获取iceberg的表信息包括列信息
     * @param tableNameList
     * @param taskParas
     * @return
     */
    private static List<IcebergTableConfig> loadIcebergTables(List<String> tableNameList, ParameterTool taskParas) {
        if(tableNameList == null || tableNameList.size() < 1){
            return null;
        }
        List<IcebergTableConfig> tableList = new ArrayList<>();
        for (String tableName : tableNameList){
            IcebergTableConfig table = new IcebergTableConfig();
            List<IcebergFieldConfig> fieldList = new ArrayList<>();
            String fieldStr = taskParas.get("iceberg.field." + tableName);
            String versionStr = taskParas.get("iceberg.version." + tableName);
            String[] fieldArray = org.apache.commons.lang3.StringUtils.split(fieldStr,",");
            for (String str : fieldArray){
                IcebergFieldConfig field = new IcebergFieldConfig();
                String[] fieldPropStr = org.apache.commons.lang3.StringUtils.split(str,":");
                field.setPos(Integer.valueOf(fieldPropStr[0]));
                field.setName(fieldPropStr[1]);
                field.setType(fieldPropStr[2]);
                if(fieldPropStr.length > 4) {
                    field.setPrecision(Integer.valueOf(fieldPropStr[3]));
                    field.setScale(Integer.valueOf(fieldPropStr[4]));
                }
                fieldList.add(field);
            }
            table.setName(tableName);
            table.setVersion(versionStr);
            table.setFieldList(fieldList);
            tableList.add(table);
        }
        return tableList;
    }
}
